package com.zhangh.summer.admin.support.ds.es;


import com.zhangh.summer.admin.support.spring.util.SpringBeanUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.core.env.Environment;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class EsClientFactory implements  IEsClientFactory{

    private static final Map<String, IEsClient> CLIENT_MAP = new ConcurrentHashMap<>();

    private static final String DEFAULT_INDEX="ES_DEFAULT_INDEX";

    private static final String DEFAULT_TYPE="ES_DEFAULT_TYPE";

    private RestClient restClient ;

    private String httpHostPort;

    private Environment env;

    public EsClientFactory(){
        this.env= SpringBeanUtil.getBeanByType(Environment.class);
    }

    public EsClientFactory(String dsName){
        this.env= SpringBeanUtil.getBeanByType(Environment.class);
        String version=env.getProperty("ds.elasticsearch."+dsName+".version","7.x");//默认7.x 走restful client
        if (version.equals("7.x")) {
            log.info("build es:{} 7.x client",dsName);
            this.restClient=this.buildRestClient(dsName);
        }else{//除了7.x版本的，都走http的请求
            log.info("build es:{} 2.x client",dsName);
            this.httpHostPort=this.buildHttpClient(dsName);
        }
    }

    public IEsClient client(String index){
        IEsClient esClient =CLIENT_MAP.get(index+DEFAULT_TYPE);
        if (null!= esClient){
            return CLIENT_MAP.get(index+DEFAULT_TYPE);
        }
        synchronized (this){
            if (null!=restClient) {
                esClient = new Es7xClient(restClient, index);
            }else{
                esClient=new Es2xClient(httpHostPort,index);
            }
            CLIENT_MAP.put(index, esClient);
        }
        return esClient;
    }

    public IEsClient client(String index, String type){
        IEsClient esClient =CLIENT_MAP.get(index+type);
        if (null!= esClient){
            return CLIENT_MAP.get(index+type);
        }
        synchronized (this){
            if (null!=restClient) {
                esClient = new Es7xClient(restClient, index);//ES7 无type
            }else{
                esClient=new Es2xClient(httpHostPort,index,type);
            }
            CLIENT_MAP.put(index, esClient);
        }
        return esClient;
    }

    public IEsClient client(){
        IEsClient esClient =CLIENT_MAP.get(DEFAULT_INDEX+DEFAULT_TYPE);
        if (null!= esClient){
            return CLIENT_MAP.get(DEFAULT_INDEX+DEFAULT_TYPE);
        }
        synchronized (this){
            if (null!=restClient) {
                esClient = new Es7xClient(restClient);
            }else{
                esClient = new Es2xClient(httpHostPort);
            }
            CLIENT_MAP.put(DEFAULT_INDEX+DEFAULT_TYPE, esClient);
        }
        return esClient;
    }

    private String buildHttpClient(String dsName){
        int port=80;
        if (StringUtils.isNotBlank(env.getProperty("ds.elasticsearch."+dsName+".port"))){
            port=Integer.valueOf(env.getProperty("ds.elasticsearch."+dsName+".port"));
        }
        String hosts=env.getProperty("ds.elasticsearch."+dsName+".host");
        String hostArray[]=hosts.split(",");
        HttpHost[] httpHostArray=new HttpHost[hostArray.length];
        for (int i = 0; i < hostArray.length; i++) {
            String hostPort=hostArray[i];
            String sHost=hostPort;
            int sPort=port;
            if (hostPort.contains(":")){
                sHost=hostPort.split(":")[0];
                sPort=Integer.valueOf(hostPort.split(":")[1]);
            }
            httpHostArray[i]=new HttpHost(sHost,sPort,"http");
        }
        String httpHostPort="http://"+httpHostArray[0].getHostName()+":"+httpHostArray[0].getPort();
        if (httpHostArray[0].getPort()==80){
            httpHostPort= "http://"+httpHostArray[0].getHostName();
        }
       //return httpHostArray[0].getAddress().getHostAddress();
        return httpHostPort;
    }

    private RestHighLevelClient buildRestHighLevelClient(String dsName){

        int port=0;
        if (StringUtils.isNotBlank(env.getProperty("ds.elasticsearch."+dsName+".port"))){
            port=Integer.valueOf(env.getProperty("ds.elasticsearch."+dsName+".port"));
        }

        String hosts=env.getProperty("ds.elasticsearch."+dsName+".host");
        String hostArray[]=hosts.split(",");
        HttpHost[] httpHostArray=new HttpHost[hostArray.length];
        for (int i = 0; i < hostArray.length; i++) {
            String hostPort=hostArray[i];
            String sHost=hostPort;
            int sPort=port;
            if (hostPort.contains(":")){
                sHost=hostPort.split(":")[0];
                sPort=Integer.valueOf(hostPort.split(":")[1]);
            }
            httpHostArray[i]=new HttpHost(sHost,sPort,"http");
        }

        String clusterName=env.getProperty("ds.elasticsearch."+dsName+".cluster.name");
        String username=env.getProperty("ds.elasticsearch."+dsName+".username");
        String password=env.getProperty("ds.elasticsearch."+dsName+".password");

        //有账号密码
        RestHighLevelClient client=null;
        if (StringUtils.isNotEmpty(username) && StringUtils.isNotEmpty(password)) {
            CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));
            client = new RestHighLevelClient(RestClient.builder(httpHostArray)
                    .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                        @Override
                        public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                            return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                        }
                    }));
        } else {
            client = new RestHighLevelClient(RestClient.builder(httpHostArray));
        }

        return client;
    }

    private RestClient buildRestClient(String dsName){

        int port=0;
        if (StringUtils.isNotBlank(env.getProperty("ds.elasticsearch."+dsName+".port"))){
            port=Integer.valueOf(env.getProperty("ds.elasticsearch."+dsName+".port"));
        }

        String hosts=env.getProperty("ds.elasticsearch."+dsName+".host");
        String hostArray[]=hosts.split(",");
        HttpHost[] httpHostArray=new HttpHost[hostArray.length];
        for (int i = 0; i < hostArray.length; i++) {
            String hostPort=hostArray[i];
            String sHost=hostPort;
            int sPort=port;
            if (hostPort.contains(":")){
                sHost=hostPort.split(":")[0];
                sPort=Integer.valueOf(hostPort.split(":")[1]);
            }
            httpHostArray[i]=new HttpHost(sHost,sPort,"http");
        }

        String clusterName=env.getProperty("ds.elasticsearch."+dsName+".cluster.name");
        String username=env.getProperty("ds.elasticsearch."+dsName+".username");
        String password=env.getProperty("ds.elasticsearch."+dsName+".password");

        //有账号密码
        RestClient client=null;
        if (StringUtils.isNotEmpty(username) && StringUtils.isNotEmpty(password)) {
            CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));
            client = RestClient.builder(httpHostArray)
                    .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                        @Override
                        public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                            return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                        }
                    }).build();
        } else {
            client = RestClient.builder(httpHostArray).build();
        }

        return client;
    }
}
